/**
 * 
 */
package xj.toolkit.fsm.core;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

import javax.management.ObjectName;

import org.apache.commons.beanutils.PropertyUtils;
import org.apache.commons.collections4.Closure;
import org.apache.commons.collections4.Predicate;
import org.apache.commons.collections4.Transformer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.jmx.export.MBeanExporter;

import xj.toolkit.ebus.EventBus;
import xj.toolkit.fsm.ContextKeyFactory;
import xj.toolkit.fsm.FSMContext;
import xj.toolkit.fsm.FSMDashboard;
import xj.toolkit.fsm.FiniteStateMachine;
import xj.toolkit.fsm.MutableFSMContext;
import xj.toolkit.fsm.State;
import xj.toolkit.fsm.StateRepository;
import xj.toolkit.fsm.annotation.IndicateInterface;
import xj.toolkit.fsm.annotation.Indicator;
import xj.toolkit.fsm.event.TimeoutEvent;

/**
 * @author hp
 * 
 */
public abstract class SmartFSM implements FiniteStateMachine {

	private static final Logger logger = LoggerFactory
			.getLogger(SmartFSM.class);

	private final Map<Object, SmartContext> contextHolder = new ConcurrentHashMap<Object, SmartContext>();

	private final Map<TimeoutEvent, Long> timeoutHolder = new HashMap<TimeoutEvent, Long>();

	private final String name;
	private final int fsmId;
	private volatile boolean running = true;
	private volatile ContextKeyFactory keyFactory;
	private volatile Predicate initeventAcceptor;
	private volatile String initStateName;
	private volatile StateRepository stateRepository;
	private final ThreadPoolExecutor workService;
	private final ScheduledThreadPoolExecutor innerService;

	private volatile EventBus eventBus;
	private volatile String eventStateChanged;
	private volatile boolean logStateChanged = true;
	private volatile String eventOfDestroyContext = null;

	private volatile int executorPoolSizeLimit = 100;
	private final AtomicInteger totalJobCount = new AtomicInteger(0);
	private final AtomicInteger activeJobCount = new AtomicInteger(0);
	private final AtomicInteger largestActiveJobCount = new AtomicInteger(0);
	private final AtomicInteger periodLargestActiveJobCount = new AtomicInteger(
			0);

	private volatile boolean checkBusy = true;
	private volatile int busyThreshold = 200;
	private volatile Closure busyHandler = null;
	private volatile Transformer snapshotFactory = null;
	private final AtomicInteger dealHandledCount = new AtomicInteger(0);
	private final AtomicInteger dealCompletedCount = new AtomicInteger(0);
	private final AtomicInteger dealBypassCount = new AtomicInteger(0);

	// 50L ms
	private long intervalOfCheckTimeoutEvent = 50L;
	private ScheduledFuture<?> futureOfCheckTimeoutEvent = null;

	// 1L s
	private long intervalOfAdjustExecutorCorePoolSize = 1 * 1000L;
	private ScheduledFuture<?> futureOfAdjustExecutorCorePoolSize = null;

	// dashboard statistics value
	private volatile long dealLargestTimeToLive = -1;
	private volatile long dealLargestTimeToActive = -1;
	private volatile long dealSmallestTimeToLive = Integer.MAX_VALUE;
	private volatile long dealSmallestTimeToActive = Integer.MAX_VALUE;
	private final AtomicLong dealTotalTimeToLive = new AtomicLong(0);
	private final AtomicLong dealTotalTimeToActive = new AtomicLong(0);

	// deal statistics value
	private volatile Transformer transformerOfTTL = null;
	private volatile Transformer transformerOfTTA = null;

	private final Map<Object, AtomicLong> erCounters = new HashMap<Object, AtomicLong>();

	private static class ERTuple {
		Transformer trans;
		Map<Object, AtomicLong> counter = new HashMap<Object, AtomicLong>();
	}

	private final Map<String, ERTuple> erTuplesOfTTL = new HashMap<String, ERTuple>();

	private final Map<String, ERTuple> erTuplesOfTTA = new HashMap<String, ERTuple>();

	private final Map<String, ERTuple> erTuplesOfStateAwareTTL = new HashMap<String, ERTuple>();

	// private volatile FSMStatisticer statisticer;

	// JMX support
	private volatile MBeanExporter mbeanExporter = null;

	// FSM JMX's prefix
	private volatile String objectNamePrefix;

	private final List<ObjectName> objectNameList = new ArrayList<ObjectName>();

	private String createObjectName(final String suffix) {
		return objectNamePrefix + "," + suffix;
	}

	private String createObjectNameForDeal(final Object reason,
			final String suffix) {
		return objectNamePrefix + ",module=deal,result=" + reason.toString()
				+ "," + suffix;
	}

	private final SmartContext.StatusListener ctxStatusListener = new SmartContext.StatusListener() {
		public void onActive(SmartContext ctx) {
			incActiveJobCount();
		}

		public void onUnactive(SmartContext ctx) {
			decActiveJobCount();
		}
	};

	private void registerMBean(final String on, final Object mbean) {
		final MBeanExporter exporter = mbeanExporter;

		if (null != exporter) {
			ObjectName objectName;
			try {
				objectName = new ObjectName(on);
				exporter.registerManagedResource(mbean, objectName);
				objectNameList.add(objectName);
			} catch (Exception e) {
				logger.error("registerMBean", e);
			}
		}

	}

	private void unregisterAllMBean() {
		final MBeanExporter exporter = mbeanExporter;

		if (null != exporter) {
			for (ObjectName on : objectNameList) {
				try {
					exporter.getServer().unregisterMBean(on);
				} catch (Exception e) {
					logger.error("unregisterAllObjectName:", e);
				}
			}
			objectNameList.clear();
		}
	}

	public SmartFSM(final String fsmName, final int fsmId) {
		this.name = fsmName;
		this.fsmId = fsmId;
		this.objectNamePrefix = "xj.toolkit:type=smartfsm,name=" + this.name
				+ ",id=" + this.fsmId;
		this.workService = new ThreadPoolExecutor(0, executorPoolSizeLimit, 1L,
				TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(),
				new ThreadFactory() {
					public Thread newThread(Runnable r) {
						return new Thread(r, name);
					}
				});

		this.innerService = new ScheduledThreadPoolExecutor(1,
				new ThreadFactory() {
					public Thread newThread(Runnable r) {
						return new Thread(r, name + "-timer");
					}
				});

		rescheduleCheckTimeoutEvent();

		rescheduleAdjustThreadPoolCoreSize();
	}

	public void start() {
		registerMBean(createObjectName("module=dashboard"), new DashboardImpl());
		registerMBean(createObjectName("module=fsm"), this);
	}

	public void stop() {
		if (!innerService.isShutdown()) {
			innerService.shutdownNow();
		}

		if (!workService.isShutdown()) {
			workService.shutdownNow();
		}

		unregisterAllMBean();

		running = false;

		if (logger.isInfoEnabled()) {
			logger.info("fsm {}/{} stopped.", this.name, this.fsmId);
		}
	}

	public boolean isRunning() {
		return running;
	}

	/**
	 * @return the eventOfDestroyContext
	 */
	public String getEventOfDestroyContext() {
		return eventOfDestroyContext;
	}

	/**
	 * @param eventOfDestroyContext
	 *            the eventOfDestroyContext to set
	 */
	public void setEventOfDestroyContext(final String eventOfDestroyContext) {
		this.eventOfDestroyContext = eventOfDestroyContext;
	}

	/**
	 * @return the initState
	 */
	public String getInitStateName() {
		return initStateName;
	}

	/**
	 * @param initState
	 *            the initState to set
	 */
	public void setInitStateName(final String initState) {
		this.initStateName = initState;
	}

	public Transformer getSnapshotFactory() {
		return snapshotFactory;
	}

	public void setSnapshotFactory(final Transformer snapshotFactory) {
		this.snapshotFactory = snapshotFactory;
	}

	public EventBus getEventBus() {
		return eventBus;
	}

	public void setEventBus(final EventBus eventBus) {
		this.eventBus = eventBus;
	}

	public String getEventStateChanged() {
		return eventStateChanged;
	}

	public void setEventStateChanged(final String eventStateChanged) {
		this.eventStateChanged = eventStateChanged;
	}

	public String getName() {
		return name;
	}

	public int getExecutorPoolSizeLimit() {
		return executorPoolSizeLimit;
	}

	public void setExecutorPoolSizeLimit(final int poolSize) {
		if (poolSize > 0) {
			this.executorPoolSizeLimit = poolSize;
			this.reducePoolSize(poolSize);
			this.workService.setMaximumPoolSize(poolSize);
		}
	}

	public String getStateRepositoryAsString() {
		return stateRepository.toString();
	}

	public StateRepository getStateRepository() {
		return stateRepository;
	}

	public void setStateRepository(final StateRepository stateRepository) {
		this.stateRepository = stateRepository;
	}

	public ContextKeyFactory getKeyFactory() {
		return keyFactory;
	}

	public void setKeyFactory(final ContextKeyFactory keyFactory) {
		this.keyFactory = keyFactory;
	}

	/**
	 * @return the initeventAcceptor
	 */
	public Predicate getIniteventAcceptor() {
		return initeventAcceptor;
	}

	/**
	 * @param initeventAcceptor
	 *            the initeventAcceptor to set
	 */
	public void setIniteventAcceptor(final Predicate initeventAcceptor) {
		this.initeventAcceptor = initeventAcceptor;
	}

	public Closure getBusyHandler() {
		return busyHandler;
	}

	public void setBusyHandler(final Closure busyHandler) {
		this.busyHandler = busyHandler;
	}

	public boolean isLogStateChanged() {
		return logStateChanged;
	}

	public void setLogStateChanged(final boolean logStateChanged) {
		this.logStateChanged = logStateChanged;
	}

	public boolean isCheckBusy() {
		return checkBusy;
	}

	public void setCheckBusy(final boolean checkBusy) {
		this.checkBusy = checkBusy;
	}

	public int getBusyThreshold() {
		return busyThreshold;
	}

	public void setBusyThreshold(final int busyThreshold) {
		this.busyThreshold = busyThreshold;
	}

	/**
	 * @return the transformerOfTimeToLive
	 */
	public Transformer getTransformerOfTimeToLive() {
		return transformerOfTTL;
	}

	/**
	 * @param transformerOfTimeToLive
	 *            the transformerOfTimeToLive to set
	 */
	public void setTransformerOfTimeToLive(Transformer transformerOfTimeToLive) {
		this.transformerOfTTL = transformerOfTimeToLive;
	}

	// public void setupTransformerOfEndReasonTimeToLive(String endReason,
	// Transformer transformerOfTimeToLive) {
	// ERTuple tuple = erTuplesOfTTL.get(endReason);
	//
	// if ( null != tuple ) {
	// throw new RuntimeException("EndReason[" + endReason
	// +"]'s transformer already setup!");
	// }
	//
	// tuple = new ERTuple();
	// tuple.trans = transformerOfTimeToLive;
	//
	// erTuplesOfTTL.put(endReason, tuple);
	// }
	//
	// public void setupTransformerOfEndReasonTimeToActive(String endReason,
	// Transformer transformerOfTimeToActive) {
	// ERTuple tuple = erTuplesOfTTA.get(endReason);
	//
	// if ( null != tuple ) {
	// throw new RuntimeException("EndReason[" + endReason
	// +"]'s transformer already setup!");
	// }
	//
	// tuple = new ERTuple();
	// tuple.trans = transformerOfTimeToActive;
	//
	// erTuplesOfTTA.put(endReason, tuple);
	// }
	//
	/**
	 * @return the transformerOfTimeToActive
	 */
	public Transformer getTransformerOfTimeToActive() {
		return transformerOfTTA;
	}

	/**
	 * @param transformerOfTimeToActive
	 *            the transformerOfTimeToActive to set
	 */
	public void setTransformerOfTimeToActive(
			Transformer transformerOfTimeToActive) {
		this.transformerOfTTA = transformerOfTimeToActive;
	}

	/**
	 * @return the fsmId
	 */
	public int getFsmId() {
		return fsmId;
	}

	/**
	 * @param unit
	 * @return
	 * @see java.util.concurrent.ThreadPoolExecutor#getKeepAliveTime(java.util.concurrent.TimeUnit)
	 */
	public long getExecutorKeepAliveTime() {
		if (null != workService) {
			return workService.getKeepAliveTime(TimeUnit.MILLISECONDS);
		} else {
			return -1;
		}
	}

	/**
	 * @param time
	 * @param unit
	 * @see java.util.concurrent.ThreadPoolExecutor#setKeepAliveTime(long,
	 *      java.util.concurrent.TimeUnit)
	 */
	public void setExecutorKeepAliveTime(long time) {
		if (null != workService) {
			workService.setKeepAliveTime(time, TimeUnit.MILLISECONDS);
		}
	}

	/**
	 * @return the intervalOfCheckTimeoutEvent
	 */
	public long getIntervalOfCheckTimeoutEvent() {
		return intervalOfCheckTimeoutEvent;
	}

	/**
	 * @param intervalOfCheckTimeoutEvent
	 *            the intervalOfCheckTimeoutEvent to set
	 */
	public synchronized void setIntervalOfCheckTimeoutEvent(
			long intervalOfCheckTimeoutEvent) {
		this.intervalOfCheckTimeoutEvent = intervalOfCheckTimeoutEvent;
		this.rescheduleCheckTimeoutEvent();
	}

	/**
	 * @return the intervalOfAdjustThreadPoolCoreSize
	 */
	public long getIntervalOfAdjustExecutorCorePoolSize() {
		return intervalOfAdjustExecutorCorePoolSize;
	}

	/**
	 * @param intervalOfAdjustThreadPoolCoreSize
	 *            the intervalOfAdjustThreadPoolCoreSize to set
	 */
	public synchronized void setIntervalOfAdjustExecutorCorePoolSize(
			long intervalOfAdjustThreadPoolCoreSize) {
		this.intervalOfAdjustExecutorCorePoolSize = intervalOfAdjustThreadPoolCoreSize;
		this.rescheduleAdjustThreadPoolCoreSize();
	}

	private static void updateLargestCount(final AtomicInteger largestCount,
			final int currentCount) {
		int cnt = largestCount.get();
		boolean modified = false;
		while (!modified && cnt < currentCount) {
			modified = largestCount.compareAndSet(cnt, currentCount);
			cnt = largestCount.get();
		}
	}

	private void enlargePoolSize(int currentPoolSize) {
		if (null != workService) {
			if (currentPoolSize > workService.getCorePoolSize()) {
				synchronized (workService) {
					// double check for more effective
					if (currentPoolSize > workService.getCorePoolSize()) {
						workService.setCorePoolSize(currentPoolSize);
					}
				}
			}
		}
	}

	private void reducePoolSize(int currentPoolSize) {
		if (null != workService) {
			if (currentPoolSize < workService.getCorePoolSize()) {
				synchronized (workService) {
					if (currentPoolSize < workService.getCorePoolSize()) {
						workService.setCorePoolSize(currentPoolSize);
					}
				}
			}
		}
	}

	private void incActiveJobCount() {
		int nowCount = activeJobCount.incrementAndGet();
		updateLargestCount(largestActiveJobCount, nowCount);
		updateLargestCount(periodLargestActiveJobCount, nowCount);
		enlargePoolSize(Math.min(nowCount, executorPoolSizeLimit));
	}

	private void decActiveJobCount() {
		activeJobCount.decrementAndGet();
	}

	private void resetLargestActiveJobCount() {
		try {
			this.periodLargestActiveJobCount.set(this.activeJobCount.get());
			reducePoolSize(periodLargestActiveJobCount.get());
		} catch (Exception e) {
			logger.error("resetLargestActiveJobCount:", e);
		}
	}

	private void rescheduleCheckTimeoutEvent() {
		try {
			if (null != futureOfCheckTimeoutEvent) {
				futureOfCheckTimeoutEvent.cancel(true);
			}
			futureOfCheckTimeoutEvent = this.innerService
					.scheduleWithFixedDelay(new Runnable() {
						public void run() {
							doCheckTimeoutEvents();
						}
					}, this.intervalOfCheckTimeoutEvent,
							this.intervalOfCheckTimeoutEvent,
							TimeUnit.MILLISECONDS);
		} catch (Exception e) {
			logger.error("rescheduleCheckTimeoutEvent:", e);
		}
	}

	private void rescheduleAdjustThreadPoolCoreSize() {
		try {
			if (null != futureOfAdjustExecutorCorePoolSize) {
				futureOfAdjustExecutorCorePoolSize.cancel(true);
			}
			futureOfAdjustExecutorCorePoolSize = this.innerService
					.scheduleWithFixedDelay(new Runnable() {
						public void run() {
							resetLargestActiveJobCount();
						}
					}, this.intervalOfAdjustExecutorCorePoolSize,
							this.intervalOfAdjustExecutorCorePoolSize,
							TimeUnit.MILLISECONDS);
		} catch (Exception e) {
			logger.error("rescheduleAdjustThreadPoolCoreSize:", e);
		}
	}

	public static class LongCounter {
		final AtomicLong atomicCounter;

		public LongCounter(AtomicLong al) {
			atomicCounter = al;
		}

		public long getDealCount() {
			return atomicCounter.get();
		}
	}

	private AtomicLong createAndRegisterEndReasonCounter(final Object reason) {
		AtomicLong al = new AtomicLong(0);
		registerMBean(createObjectNameForDeal(reason, "category=all"),
				new LongCounter(al));

		return al;
	}

	private void countDealForEndReason(final Object reason) {
		String key = reason.toString();
		AtomicLong counter = erCounters.get(key);
		if (null == counter) {
			counter = createAndRegisterEndReasonCounter(reason);
			erCounters.put(key, counter);
		}
		counter.incrementAndGet();
	}

	private static Class<?> getIndicateType(final Class<?> cls) {
		final IndicateInterface anno = cls
				.getAnnotation(IndicateInterface.class);
		if (null != anno) {
			return anno.type();
		}

		return null;
	}

	private class CountersGetterImpl implements InvocationHandler {

		private final Map<Object, AtomicLong> counters;

		public CountersGetterImpl(final Map<Object, AtomicLong> counters) {
			this.counters = counters;
		}

		public Object invoke(final Object proxy, final Method method,
				final Object[] args) throws Throwable {
			final String methodName = method.getName();
			final Class<?> returnType = method.getReturnType();

			/*
			 * Inexplicably, InvocationHandler specifies that args is null when
			 * the method takes no arguments rather than a zero-length array.
			 */
			final int nargs = (args == null) ? 0 : args.length;

			if (methodName.startsWith("get")
					&& methodName.length() > 3
					&& nargs == 0
					&& (returnType.equals(long.class) || returnType
							.equals(Long.class))) {

				final String key;

				Indicator indicator = method.getAnnotation(Indicator.class);
				if (null != indicator) {
					key = indicator.key();
				} else {
					key = methodName.substring(3);
				}

				try {
					return innerService.submit(new Callable<Long>() {
						public Long call() throws Exception {
							AtomicLong ret = counters.get(key);
							return null != ret ? ret.get() : 0;
						}
					}).get();
				} catch (Exception e) {
					logger.error("DealTimeInvocationHandler.invoke:", e);
					return 0;
				}
			}

			logger.warn("Invalid method {} while valid is : getXXX"
					+ "\r\n Or invalid parameters count {}", methodName, nargs);
			return 0;
		}

	}

	private ERTuple createAndRegisterEndReasonTuple(
			final Transformer transformer, final Object reason,
			final String category) {
		final ERTuple tuple = new ERTuple();
		tuple.trans = transformer;
		Class<?> cls = getIndicateType(tuple.trans.getClass());
		if (null != cls) {
			registerMBean(
					createObjectNameForDeal(reason, "category=" + category),
					Proxy.newProxyInstance(cls.getClassLoader(),
							new Class<?>[] { cls }, new CountersGetterImpl(
									tuple.counter)));
		} else {
			logger.error("can't get IndicateType for ["
					+ tuple.trans.getClass() + "]");
		}

		return tuple;
	}

	// TODO
	private void countTimeForEndReason(final Transformer defaultTransformer,
			final Map<String, Long> stateLiveTime) {
		for (String stateName : stateLiveTime.keySet()) {
			ERTuple tuple = erTuplesOfStateAwareTTL.get(stateName);
			if (tuple == null) {
				tuple = createAndRegisterEndReasonTuple(defaultTransformer,
						"stateAwareTTL,state=" + stateName, "ttl");
				erTuplesOfStateAwareTTL.put(stateName, tuple);
			}

			final Transformer trans = tuple.trans;
			if (null != trans) {
				final Object key = trans
						.transform(stateLiveTime.get(stateName));
				if (null != key) {
					AtomicLong count = tuple.counter.get(key);
					if (null == count) {
						count = new AtomicLong(0);
						tuple.counter.put(key, count);
					}
					count.incrementAndGet();
				}
			}
		}
		stateLiveTime.clear();
	}

	private void countTimeForEndReason(long value, Object reason,
			Transformer defaultTransformer, Map<String, ERTuple> tuples,
			String category) {

		String keyOfTuple = reason.toString();
		ERTuple tuple = tuples.get(keyOfTuple);

		if (null == tuple) {
			tuple = createAndRegisterEndReasonTuple(defaultTransformer, reason,
					category);
			tuples.put(keyOfTuple, tuple);
		}

		Transformer trans = tuple.trans;
		if (null != trans) {
			final Object key = trans.transform(value);
			if (null != key) {
				AtomicLong count = tuple.counter.get(key);
				if (null == count) {
					count = new AtomicLong(0);
					tuple.counter.put(key, count);
				}
				count.incrementAndGet();
			}
		}
	}

	private void doCountDeal(Object reason, long timeToLive, long timeToActive,
			Map<String, Long> stateLiveTime) {

		if (null == reason) {
			reason = "";
		}

		countDealForEndReason(reason);

		// ttl
		countTimeForEndReason(timeToLive, reason, transformerOfTTL,
				erTuplesOfTTL, "ttl");

		// tta
		countTimeForEndReason(timeToActive, reason, transformerOfTTA,
				erTuplesOfTTA, "tta");

		// stateAware ttl
		countTimeForEndReason(transformerOfTTA, stateLiveTime);
	}

	private void doResetDealTimeCounter() {
		// dealTimeToLiveCounter.clear();
		// dealTimeToActiveCounter.clear();
		// TODO
		// clear endreason counter
	}

	private String doGetDealTimeCountAsString() {
		StringBuilder sb = new StringBuilder();

		sb.append("timeToLive:\r\n");
		// for (Map.Entry<Object, AtomicLong> entry :
		// dealTimeToLiveCounter.entrySet()) {
		// sb.append(entry.getKey());
		// sb.append(":\t");
		// sb.append(entry.getValue().get());
		// sb.append(" \r\n");
		// }
		//
		// sb.append("timeToActive:\r\n");
		// for (Map.Entry<Object, AtomicLong> entry :
		// dealTimeToActiveCounter.entrySet()) {
		// sb.append(entry.getKey());
		// sb.append(":\t");
		// sb.append(entry.getValue().get());
		// sb.append(" \r\n");
		// }

		return sb.toString();
	}

	private void doResetDealEndReasonCounter() {
		// erCounters.clear();
	}

	private String doGetDealEndReasonCountAsString() {
		final StringBuilder sb = new StringBuilder();

		for (Map.Entry<Object, AtomicLong> entry : erCounters.entrySet()) {
			sb.append(entry.getKey());
			sb.append(":\t");
			sb.append(entry.getValue().get());
			sb.append(" \r\n");
		}

		return sb.toString();
	}

	private void updateAllDealStatistics(final SmartContext ctx) {
		final long timeToLive = ctx.getTimeToLive();
		final long timeToActive = ctx.getTimeToActive();

		if (timeToLive > this.dealLargestTimeToLive) {
			this.dealLargestTimeToLive = timeToLive;
		}
		if (timeToLive < this.dealSmallestTimeToLive) {
			this.dealSmallestTimeToLive = timeToLive;
		}

		if (timeToActive > this.dealLargestTimeToActive) {
			this.dealLargestTimeToActive = timeToActive;
		}
		if (timeToActive < this.dealSmallestTimeToActive) {
			this.dealSmallestTimeToActive = timeToActive;
		}

		dealTotalTimeToLive.addAndGet(timeToLive);
		dealTotalTimeToActive.addAndGet(timeToActive);
		final Object reason = ctx.getEndReason();

		final Map<String, Long> stateLiveTime = ctx.getStateLiveTime();

		innerService.submit(new Runnable() {
			public void run() {
				doCountDeal(reason, timeToLive, timeToActive, stateLiveTime);
			}
		});
	}

	private void removeContext(final SmartContext ctx) {
		final Object key = ctx.getKey();

		this.keyFactory.removeKey(key);
		if (null != this.contextHolder.remove(key)) {
			// 移除操作有效
			this.totalJobCount.decrementAndGet();
		}

		try {
			final List<Object> pendings = ctx.destroy();
			if (null != pendings) {
				// ctx 已经结束，但还存在未处理的event
				// 将其重新放入FSM中，触发相关处理流程
				if (logger.isInfoEnabled()) {
					logger.info(
							"ctx {} destroy and {} events left, try re-accept by fsm {}",
							new Object[] { ctx, pendings.size(), this });
				}
				for (Object event : pendings) {
					if (logger.isDebugEnabled()) {
						logger.debug("fsm {} re-accept event {}", this, event);
					}
					this.acceptEvent(event);
				}
			}
		} catch (Exception e) {
			logger.error("[" + name + "]removeContext:", e);
		}

		dealCompletedCount.incrementAndGet();
		updateAllDealStatistics(ctx);
	}

	private State getStateOf(String stateName) {
		return this.stateRepository.find(stateName);
	}

	private void fireStateChanged(final FSMContext ctx, final String fromState,
			final String toState, final Object event) {
		if (logStateChanged && null != eventBus && null != eventStateChanged) {
			final Object snapshot = ctx.getSnapshot();

			if (null != snapshot) {
				try {
					PropertyUtils.setProperty(snapshot, "fsm", name);
					PropertyUtils.setProperty(snapshot, "fromState", fromState);
					PropertyUtils.setProperty(snapshot, "toState", toState);
					PropertyUtils.setProperty(snapshot, "eventRecved",
							null != event ? event.toString() : "");
					PropertyUtils.setProperty(snapshot, "time", new Date());
				} catch (IllegalAccessException e) {
					logger.error("fireStateChanged:", e);
				} catch (InvocationTargetException e) {
					logger.error("fireStateChanged:", e);
				} catch (NoSuchMethodException e) {
					logger.error("fireStateChanged:", e);
				}

				eventBus.fireEvent(this.eventStateChanged, snapshot);
			}
		}
	}

	private void initSnapshot(final MutableFSMContext ctx) {
		if (this.logStateChanged && (null != this.snapshotFactory)) {
			try {
				ctx.setSnapshot(snapshotFactory.transform(ctx));
			} catch (Exception e) {
				logger.error("[" + name + "]initSnapshot:", e);
			}
		}
	}

	private void initContext(final SmartContext ctx, final Object event) {
		final State state = getStateOf(ctx.getCurrentState());
		if (null == state) {
			logger.warn("Can not found valid state for ["
					+ ctx.getCurrentState() + "] , remove context [" + ctx
					+ "]");
			removeContext(ctx);
		} else {

			initSnapshot(ctx);

			boolean ret = false;

			try {
				ret = state.enter(this, ctx);
			} catch (Exception e) {
				logger.error(ctx.getCurrentState() + ".enter:", e);
			}

			if (!ret) {
				if (logger.isDebugEnabled()) {
					logger.debug("enter return false, ctx [" + ctx
							+ "] will end normally.");
				}
				removeContext(ctx);
				return;
			}
			fireStateChanged(ctx, "", ctx.getCurrentState(), event);

			initSnapshot(ctx);

			dispatchEvent(ctx, event);
		}
	}

	private void dispatchEvent(final SmartContext ctx, final Object event) {
		final long stateBegin = System.currentTimeMillis();
		final State currentState = getStateOf(ctx.getCurrentState());
		if (null == currentState) {
			logger.error("Internal Error: Can not found valid state for ["
					+ ctx.getCurrentState() + "] , remove context [" + ctx
					+ "]");
			removeContext(ctx);
		}

		String nextStateName = null;

		try {
			nextStateName = currentState.acceptEvent(this, ctx, event);
		} catch (Exception e) {
			logger.error(ctx.getCurrentState() + ".acceptEvent:", e);
		}

		if (null == nextStateName) {
			// this context's FSM state reach end
			try {
				currentState.leave(this, ctx);
			} catch (Exception e) {
				logger.error(ctx.getCurrentState() + ".leave:", e);
			}

			if (logger.isDebugEnabled()) {
				logger.debug("ctx [" + ctx + "] will end normally.");
			}

			fireStateChanged(ctx, ctx.getCurrentState(), "", event);
			removeContext(ctx);
		} else if (nextStateName.equals(ctx.getCurrentState())) {
			// no change
		} else {
			final State nextState = getStateOf(nextStateName);
			if (null == nextState) {
				logger.error("Internal Error: Can not found valid state for ["
						+ nextStateName + "] , just ignore");
			} else {
				// next state is valid
				try {
					currentState.leave(this, ctx);
				} catch (Exception e) {
					logger.error(ctx.getCurrentState() + ".leave:", e);
				}
				// TODO 上个状态完毕,进入下个状态
				if (this.getInitStateName().equalsIgnoreCase(
						ctx.getCurrentState())) {
					ctx.setStateLiveTime(ctx.getCurrentState(),
							System.currentTimeMillis() - stateBegin);
				} else {
					ctx.setStateLiveTime(ctx.getCurrentState(),
							System.currentTimeMillis() - ctx.getLastModify());
				}

				final String prevStateName = ctx.getCurrentState();
				ctx.setCurrentState(nextStateName);

				boolean ret = false;
				try {
					ret = nextState.enter(this, ctx);
				} catch (Exception e) {
					logger.error(ctx.getCurrentState() + ".enter:", e);
				}

				fireStateChanged(ctx, prevStateName, ctx.getCurrentState(),
						event);
				initSnapshot(ctx);

				if (!ret) {
					if (logger.isDebugEnabled()) {
						logger.debug("enter return false, ctx [" + ctx
								+ "] will end normally.");
					}

					removeContext(ctx);
				}
			}
		}

		try {
			ctx.endOfDispatchEvent(new Closure() {

				public void execute(final Object input) {
					dispatchEvent(ctx, input);
				}
			});
		} catch (Exception e) {
			logger.error("ctx.endOfDispatchEvent{}, try end ctx", e);
			removeContext(ctx);
		}
	}

	private void incDealBypass() {
		dealBypassCount.incrementAndGet();
	}

	private void incDealHandledStart() {
		dealHandledCount.incrementAndGet();
	}

	private boolean checkIfExceedLimit(Object event) {
		if (this.checkBusy && this.activeJobCount.get() >= this.busyThreshold) {

			incDealBypass();

			// 超过CTX限制值
			if (null != busyHandler) {
				try {
					busyHandler.execute(event);
				} catch (Exception e) {
					logger.error(busyHandler.toString() + ".execute:", e);
				}
			} else {
				logger.error("reach active ctx limit, and missing BusyHandler.");
			}
			return true;
		}
		return false;
	}

	public void acceptEvent(final Object event) {
		final Object key = keyFactory.transformToKey(event);

		if (null == key) {
			logger.warn("event " + event
					+ " can not transform key, just ignore");
			return;
		}
		// lookup key in contextHolder
		final SmartContext ctx = contextHolder.get(key);
		if (null == ctx) {
			if (!this.initeventAcceptor.evaluate(event)) {
				if (logger.isDebugEnabled()) {
					logger.debug("event [" + event
							+ "] not acceptable, just ignore");
				}
				return;
			}

			if (checkIfExceedLimit(event)) {
				if (logger.isDebugEnabled()) {
					logger.debug("FSM[" + name + "] busy, event[" + event
							+ "] has been dropped.");
				}
				return;
			}

			final SmartContext initCtx = createSmartContext();
			if (null == initCtx) {
				logger.warn("Can not create context for event " + event
						+ " , just ignore");
				return;
			}

			incDealHandledStart();

			// init ctx, and continue
			initCtx.setService(workService);
			initCtx.setKey(key);
			initCtx.setActived();

			if (null == contextHolder.put(key, initCtx)) {
				// add new context
				totalJobCount.incrementAndGet();
			}

			workService.submit(new Runnable() {

				public void run() {
					initContext(initCtx, event);
				}
			});
		} else {
			try {
				ctx.processEvent(event, new Closure() {

					public void execute(final Object input) {
						dispatchEvent(ctx, input);
					}
				});
			} catch (Exception e) {
				logger.error("ctx.processEvent error, try end ctx", e);
				removeContext(ctx);
			}
		}
	}

	// in timerExecService thread
	private void doScheduleTimeout(final TimeoutEvent event, final long delay) {
		timeoutHolder.put(event, System.currentTimeMillis() + delay);
	}

	// in timerExecService thread
	private void doCancelTimeout(TimeoutEvent event) {
		if (null == timeoutHolder.remove(event)) {
			if (logger.isWarnEnabled()) {
				logger.warn("Can not found matched timeout event[" + event
						+ "], maybe has been emited.");
			}
		} else {
			if (logger.isDebugEnabled()) {
				logger.debug("cancel timeout event[" + event + "] succeed.");
			}
		}
	}

	// in timerExecService thread
	private void doCheckTimeoutEvents() {
		final long now = System.currentTimeMillis();

		try {
			final Set<Map.Entry<TimeoutEvent, Long>> entrySet = timeoutHolder
					.entrySet();

			final Iterator<Map.Entry<TimeoutEvent, Long>> iter = entrySet
					.iterator();
			while (iter.hasNext()) {
				final Map.Entry<TimeoutEvent, Long> entry = iter.next();
				if (entry.getValue() <= now) {
					// condition matched, emit timeout event
					iter.remove();
					this.acceptEvent(entry.getKey());
				}
			}
		} catch (Exception e) {
			logger.error("doCheckTimeoutEvents:", e);
		}
	}

	public Runnable scheduleTimeout(final TimeoutEvent event, final long delay) {
		this.innerService.submit(new Runnable() {

			public void run() {
				doScheduleTimeout(event, delay);
			}
		});

		return new Runnable() {

			public void run() {
				innerService.submit(new Runnable() {

					public void run() {
						doCancelTimeout(event);
					}
				});
			}
		};
	}

	public class DashboardImpl implements FSMDashboard {
		public long getDealAverageTimeToActive() {
			int completed = dealCompletedCount.get();
			return completed > 0 ? dealTotalTimeToActive.get() / completed : -1;
		}

		public long getDealAverageTimeToLive() {
			int completed = dealCompletedCount.get();
			return completed > 0 ? dealTotalTimeToLive.get() / completed : -1;
		}

		public long getDealBypassCount() {
			return dealBypassCount.get();
		}

		public long getDealCompletedCount() {
			return dealCompletedCount.get();
		}

		public long getDealHandledCount() {
			return dealHandledCount.get();
		}

		public long getDealLargestTimeToActive() {
			return dealLargestTimeToActive;
		}

		public long getDealLargestTimeToLive() {
			return dealLargestTimeToLive;
		}

		public long getDealSmallestTimeToActive() {
			return dealSmallestTimeToActive;
		}

		public long getDealSmallestTimeToLive() {
			return dealSmallestTimeToLive;
		}

		public int getExecutorActiveThreadCount() {
			return workService.getActiveCount();
		}

		public long getExecutorCompletedTaskCount() {
			return workService.getCompletedTaskCount();
		}

		public int getExecutorCorePoolSize() {
			return workService.getCorePoolSize();
		}

		public int getExecutorCurrentPoolSize() {
			return workService.getPoolSize();
		}

		public long getExecutorHandledTaskCount() {
			return workService.getTaskCount();
		}

		public int getExecutorLargestPoolSize() {
			return workService.getLargestPoolSize();
		}

		public int getExecutorMaximumPoolSize() {
			return workService.getMaximumPoolSize();
		}

		public int getExecutorPendingTaskCount() {
			if (null != workService) {
				return workService.getQueue().size();
			}
			return -1;
		}

		public int getJobActiveCount() {
			return activeJobCount.get();
		}

		public int getJobInactiveCount() {
			return getJobTotalCount() - getJobActiveCount();
		}

		public int getJobTotalCount() {
			return totalJobCount.get();
		}

		public int getJobLargestActiveCount() {
			return largestActiveJobCount.get();
		}

		public int getJobPeriodLargestActiveCount() {
			return periodLargestActiveJobCount.get();
		}

		public int getJobTimeoutEventCount() {
			return timeoutHolder.size();
		}

		public String[] getJobsDetail() {
			final Set<Map.Entry<Object, SmartContext>> ctxs = contextHolder
					.entrySet();

			final List<String> ret = new ArrayList<String>();
			final SimpleDateFormat formater = new SimpleDateFormat(
					"yyMMdd-HH:mm:ss.SSS");

			for (Map.Entry<Object, SmartContext> entry : ctxs) {
				final StringBuilder sb = new StringBuilder();

				sb.append(entry.getKey().toString());
				sb.append(":[");

				FSMContext ctx = entry.getValue();
				sb.append(ctx.getCurrentState());
				sb.append("]create:");
				sb.append(formater.format(new Date(ctx.getCreateTime())));
				sb.append("/last:");
				sb.append(formater.format(new Date(ctx.getLastModify())));
				ret.add(sb.toString());
			}

			return ret.toArray(new String[0]);
		}

		public int getInnerServiceTaskCount() {
			if (null != innerService) {
				return innerService.getQueue().size();
			}
			return -1;
		}

		public void resetDealTimeExtremum() {
			dealLargestTimeToLive = -1;
			dealLargestTimeToActive = -1;
			dealSmallestTimeToLive = Integer.MAX_VALUE;
			dealSmallestTimeToActive = Integer.MAX_VALUE;
		}

		public String getDealTimeCountAsString() {
			try {
				return innerService.submit(new Callable<String>() {
					public String call() throws Exception {
						return doGetDealTimeCountAsString();
					}
				}).get();
			} catch (Exception e) {
				logger.error("Dashboard.getDealTimeCountAsString:", e);
				return null;
			}
		}

		public void resetDealTimeCounter() {
			try {
				innerService.submit(new Runnable() {
					public void run() {
						doResetDealTimeCounter();
					}
				}).get();
			} catch (Exception e) {
				logger.error("Dashboard.resetDealTimeCounter:", e);
			}
		}

		public String getDealEndReasonCountAsString() {
			try {
				return innerService.submit(new Callable<String>() {
					public String call() throws Exception {
						return doGetDealEndReasonCountAsString();
					}
				}).get();
			} catch (Exception e) {
				logger.error("Dashboard.getDealEndReasonCountAsString:", e);
				return null;
			}
		}

		public void resetDealEndReasonCounter() {
			try {
				innerService.submit(new Runnable() {
					public void run() {
						doResetDealEndReasonCounter();
					}
				}).get();
			} catch (Exception e) {
				logger.error("Dashboard.resetDealEndReasonCounter:", e);
			}
		}
	}

	public FSMDashboard createDashboard() {
		return new DashboardImpl();
	}

	/**
	 * @return the mbeanExport
	 */
	public MBeanExporter getMbeanExporter() {
		return mbeanExporter;
	}

	/**
	 * @param mbeanExport
	 *            the mbeanExport to set
	 */
	public void setMbeanExporter(MBeanExporter mbeanExport) {
		this.mbeanExporter = mbeanExport;
	}

	/**
	 * @return the objectNamePrefix
	 */
	public String getObjectNamePrefix() {
		return objectNamePrefix;
	}

	/**
	 * @param objectNamePrefix
	 *            the objectNamePrefix to set
	 */
	public void setObjectNamePrefix(final String objectNamePrefix) {
		this.objectNamePrefix = objectNamePrefix;
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see java.lang.Object#toString()
	 */
	@Override
	public String toString() {
		return this.name;
	}

	protected abstract SmartContext createFSMContext();

	private SmartContext createSmartContext() {
		final SmartContext ctx = createFSMContext();

		ctx.setCurrentState(this.initStateName);
		if (null != eventOfDestroyContext) {
			ctx.setEventOfDestroy(eventOfDestroyContext);
		}
		ctx.setStatusListener(ctxStatusListener);

		return ctx;
	}

}
